package drds.binlog.parse.mysql;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import drds.binlog.binlog_event.LogEvent;
import drds.binlog.common.Authors;
import drds.binlog.parse.exception.ParseException;
import drds.common.Author;

@Author(name = Authors.LI_YANG)
class SinkStoreStage implements EventHandler<MessageEvent>, LifecycleAware
{

    private MultiStageCoprocessor multiStageCoprocessor;

    public SinkStoreStage(MultiStageCoprocessor multiStageCoprocessor)
    {
        this.multiStageCoprocessor = multiStageCoprocessor;
    }

    public void onEvent(MessageEvent messageEvent, long sequence, boolean endOfBatch) throws Exception
    {
        try
        {
            if (messageEvent.getEntry() != null)
            {
                multiStageCoprocessor.entryListTransaction.add(messageEvent.getEntry());
            }

            LogEvent logEvent = messageEvent.getLogEvent();
            if (multiStageCoprocessor.connection instanceof Connection && logEvent.getSemi() == 1)
            {
                // semi ack回报
                ((Connection) multiStageCoprocessor.connection).sendSemiAck(logEvent.getLogHeader().getLogFileName(),
                        logEvent.getLogHeader().getLogPosition());
            }

            // removeEventExecuteDateTimeLongFromQueue for gc
            messageEvent.setLogByteBlock(null);
            messageEvent.setLogEvent(null);
            messageEvent.setTableMetaData(null);
            messageEvent.setEntry(null);
            messageEvent.setNeedDmlParse(false);
        } catch (Throwable e)
        {
            multiStageCoprocessor.parseException = new ParseException(e);
            throw multiStageCoprocessor.parseException;
        }
    }

    @Override
    public void onStart()
    {

    }

    @Override
    public void onShutdown()
    {

    }
}
